package main

import (
	"context"
	"gitee.com/ziEgate/mq/rabbitmq"
	"log"
	"os"
	"os/signal"
	"sync"
	"time"
)

func main() {
	errorChan := make(chan error, 128)

	dialOpt := &rabbitmq.DialOption{
		Username: "guest",
		Password: "guest",
		Host:     "localhost",
		Port:     "5672",
		VHost:    "/mes",
	}

	eqToSysExc := &rabbitmq.ExchangeOption{
		// Name:       "SDK.MES.DEV.FAB.EQtoMES.EXC",
		Name:       "SDK.SIHONG.SPC.PRD.FAB.EQtoSPC.EXC",
		Kind:       "topic",
		AutoDelete: true,
		Durable:    true,
		NoWait:     false,
		Args:       nil,
	}

	sysToEQExc := &rabbitmq.ExchangeOption{
		// Name:       "SDK.MES.DEV.FAB.EQtoMES.EXC",
		Name:       "SDK.SIHONG.SPC.PRD.FAB.SPCtoEQ.EXC",
		Kind:       "topic",
		AutoDelete: true,
		Durable:    true,
		NoWait:     false,
		Args:       nil,
	}

	eqReplyToSysQue := &rabbitmq.QueueOption{
		// Name:       "SDK.MES.DEV.FAB.EQreplytoMES.QUE",
		Name:       "SDK.SIHONG.SPC.PRD.FAB.EQreplytoSPC.QUE",
		AutoDelete: true,
		Durable:    true,
		Exclusive:  false,
		NoWait:     false,
		Args:       nil,
	}

	eqToSysQue := &rabbitmq.QueueOption{
		// Name:       "SDK.MES.DEV.FAB.EQtoMES.QUE",
		Name:       "SDK.SIHONG.SPC.PRD.FAB.EQtoSPC.QUE",
		AutoDelete: true,
		Durable:    true,
		Exclusive:  false,
		NoWait:     false,
		Args:       nil,
	}

	// "SDK.MES.DEV.FAB.EQtoMES.*",
	eqReplyToSysKey := "SDK.SIHONG.SPC.PRD.FAB.EQreplytoSPC.KEY"

	eqToSysKey := "SDK.SIHONG.SPC.PRD.FAB.EQtoSPC.KEY"

	// "SDK.MES.DEV.FAB.MEStoEQ.*",
	sysToEQKey := "SDK.SIHONG.SPC.PRD.FAB.SPCtoEQ.KEY"

	producer, err := rabbitmq.NewTopicProducer(
		dialOpt,
		sysToEQExc,
		eqToSysExc,
		eqReplyToSysQue,
		eqReplyToSysKey,
		"SysReplyConsumer",
		errorChan,
	)
	if err != nil {
		log.Fatal(err)
	}

	defer func() {
		if err := producer.Close(); err != nil {
			log.Fatal(err)
		}
	}()

	consumer, err := rabbitmq.NewTopicConsumer(
		dialOpt,
		eqToSysExc, sysToEQExc,
		&rabbitmq.ConsumerOption{
			Name:      "SYS",
			AutoAck:   false,
			Exclusive: false,
			NoWait:    false,
			Args:      nil,
		},
		errorChan,
	)
	if err != nil {
		log.Fatal(err)
	}

	msgs, err := consumer.Recv(
		eqToSysQue,
		eqToSysKey,
	)
	if err != nil {
		log.Fatal(err)
	}

	go func() {
		time.Sleep(5 * time.Second)

		var wg sync.WaitGroup
		wg.Add(10)
		t := time.Now()
		for i := 0; i < 10; i++ {
			go func() {
				re, err := producer.SendRequest(context.Background(), sysToEQKey, false, false, []byte("I'm SYS"), 10*time.Second)
				if err != nil {
					log.Println(err)

					return
				}

				log.Println("Reply", string(re))

				wg.Done()
			}()
		}
		wg.Wait()

		log.Println(time.Now().Sub(t))
	}()

	c := make(chan os.Signal)
	signal.Notify(c, os.Interrupt)

	for {
		select {
		case <-c:
			os.Exit(0)
		case e := <-errorChan:
			log.Println(e)
		case m := <-msgs:
			go func() {
				if m.ReplyTo != "" {
					ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
					err = consumer.SendReply(ctx, m, false, false, []byte("Hi, EQ"), 10*time.Second)
					if err != nil {
						log.Fatal(err)
					}
					cancel()
					log.Println(string(m.Body))
				} else {
					log.Println(string(m.Body))
				}
			}()
		}
	}

}
